package cn.doitedu.data_mock;

import org.apache.commons.codec.digest.Md5Crypt;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;

public class VideoDataMock {

    public static void main(String[] args) {

        for(int i=0;i<=1000;i++) {
            new Thread(new MyRunner((i+1)*1000,(i+2)*1000)).start();
        }

    }



    public static class MyRunner implements Runnable{

        int id1;
        int id2;
        KafkaProducer<String, String> producer;
        public MyRunner(int id1,int id2){
            this.id1 = id1;
            this.id2 = id2;

            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"doitedu:9092");
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.setProperty(ProducerConfig.ACKS_CONFIG,"1");

            producer = new KafkaProducer<String, String>(properties);

        }

        @Override
        public void run() {

            for(int i=0;i<1000;i++){

                int user_id = RandomUtils.nextInt(id1,id2);
                int videoId = RandomUtils.nextInt(1,100);

                String playId = Md5Crypt.md5Crypt(UUID.randomUUID().toString().getBytes()).substring(3,10);

                long action_time = System.currentTimeMillis();
                String data_play = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_play\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                //System.out.println(data_play);
                producer.send(new ProducerRecord<>("dwd-events", data_play));

                for(int j=0;j<RandomUtils.nextInt(1,10);j++){
                    action_time = System.currentTimeMillis();
                    String data_hb = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_hb\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                    //System.out.println(data_hb);
                    producer.send(new ProducerRecord<>("dwd-events", data_hb));
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }

                action_time = System.currentTimeMillis();
                String data_pause = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_pause\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                //System.out.println(data_pause);
                producer.send(new ProducerRecord<>("dwd-events", data_pause));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                action_time = System.currentTimeMillis();
                String data_resume = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_resume\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                //System.out.println(data_resume);
                producer.send(new ProducerRecord<>("dwd-events", data_resume));
                try {
                    Thread.sleep(RandomUtils.nextInt(1000,6000));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                for(int j=0;j<RandomUtils.nextInt(1,10);j++){
                    action_time = System.currentTimeMillis();
                    String data_hb = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_hb\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                    //System.out.println(data_hb);
                    producer.send(new ProducerRecord<>("dwd-events", data_hb));
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }


                action_time = System.currentTimeMillis();
                String data_stop = "{\"page_url\":\"/aa/bb/cc\",\"user_id\":"+user_id+",\"event_id\":\"video_stop\",\"action_time\":"+action_time+",\"release_channel\":\"华为应用市场\",\"device_type\":\"mi8\",\"properties\":{\"url\":\"page006\",\"video_id\":"+videoId+",\"play_id\":\""+playId+"\"}}";
                //System.out.println(data_stop);
                producer.send(new ProducerRecord<>("dwd-events", data_stop));
                producer.flush();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
